package com.steve.bigdata.flinkmonitor.utils;

import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;

import java.time.Duration;

/**
 * @Author: STEVE
 * @Description: 自定义Watermark生成器，组合了以下特性：
 * 1. 允许固定时间的乱序事件（BoundedOutOfOrderness）
 * 2. 检测并处理空闲数据源（WithIdleness）
 * @since: 2025-06-27
 */
public class BoundedOutOfOrdernessWithIdleness<T> implements WatermarkGenerator<T> {

    private final long maxOutOfOrdernessMillis; // 最大允许乱序时间（毫秒）
    private final long idleTimeoutMillis;       // 空闲超时时间（毫秒）
    private long currentMaxTimestamp;          // 当前观察到的最大事件时间
    private long lastRecordTime;              // 最后收到记录的系统时间

    /**
     * 构造函数
     *
     * @param maxOutOfOrderness 允许的最大乱序时间（如5秒）
     * @param idleTimeout       判定数据源空闲的超时时间（如1分钟）
     */
    public BoundedOutOfOrdernessWithIdleness(Duration maxOutOfOrderness, Duration idleTimeout) {
        this.maxOutOfOrdernessMillis = maxOutOfOrderness.toMillis();
        this.idleTimeoutMillis = idleTimeout.toMillis();

        // 初始化为最小可能值 + 允许的乱序时间
        this.currentMaxTimestamp = Long.MIN_VALUE + maxOutOfOrdernessMillis + 1;
        this.lastRecordTime = System.currentTimeMillis();
    }

    /**
     * 处理每个事件时调用
     *
     * @param event          当前处理的事件
     * @param eventTimestamp 从事件中提取的事件时间
     * @param output         Watermark输出器
     */
    @Override
    public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
        // 更新观察到的最大事件时间
        currentMaxTimestamp = Math.max(currentMaxTimestamp, eventTimestamp);
        // 刷新最后活动时间
        lastRecordTime = System.currentTimeMillis();

        /* 注意：这里不立即生成Watermark，而是等待周期性生成
         * 优点：减少Watermark生成开销
         * 缺点：会有一定延迟（默认200ms间隔）
         */
    }

    /**
     * 周期性生成Watermark（默认200ms调用一次）
     *
     * @param output Watermark输出器
     */
    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        // 检查是否空闲超时
        if (System.currentTimeMillis() - lastRecordTime > idleTimeoutMillis) {
            // 情况1：检测到空闲，发出最大可能Watermark推进时间
            output.emitWatermark(new Watermark(Long.MAX_VALUE));
        } else {
            // 情况2：正常情况，计算允许延迟后的Watermark
            long watermarkTimestamp = currentMaxTimestamp - maxOutOfOrdernessMillis - 1;
            output.emitWatermark(new Watermark(watermarkTimestamp));
        }
    }

}
